package cn.zhage.aimin.canal.client.handler.impl;

import cn.zhage.aimin.canal.client.EntityHandlerHolder;
import cn.zhage.aimin.canal.client.handler.EntityHandler;
import cn.zhage.aimin.canal.client.handler.MessageHandler;
import cn.zhage.aimin.canal.client.handler.RowDataHandler;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import com.google.protobuf.InvalidProtocolBufferException;

import java.util.List;

/*
 * 
 * @author 渣哥
*/
public class MessageHandlerImpl implements MessageHandler<Message> {


    private final EntityHandlerHolder entityHandlerHolder;

    private final RowDataHandler<CanalEntry.RowData> rowDataHandler;

    public MessageHandlerImpl(EntityHandlerHolder entityHandlerHolder, RowDataHandler<CanalEntry.RowData> rowDataHandler) {
        this.entityHandlerHolder = entityHandlerHolder;
        this.rowDataHandler = rowDataHandler;
    }

    @Override
    public void handle(Message message) {
        long id = message.getId();
        int size = message.getEntries().size();
        if(id != -1 || size > 0){
            List<CanalEntry.Entry> entries = message.getEntries();
            for(CanalEntry.Entry entry : entries){
                if(entry.getEntryType() != CanalEntry.EntryType.TRANSACTIONBEGIN &&
                        entry.getEntryType() != CanalEntry.EntryType.TRANSACTIONEND){
                    String tableName = entry.getHeader().getTableName();
                    EntityHandler<?> entityHandler = entityHandlerHolder.getEntityHandler(tableName);
                    CanalEntry.RowChange rowChange;
                    try {
                        rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
                    } catch (InvalidProtocolBufferException e) {
                        throw new RuntimeException(e);
                    }
                    CanalEntry.EventType eventType = rowChange.getEventType();
                    List<CanalEntry.RowData> rowDatasList = rowChange.getRowDatasList();
                    for(CanalEntry.RowData rowData : rowDatasList){
                        rowDataHandler.handle(rowData,eventType,entityHandler);
                    }
                }
            }
        }
    }
}
